package top.ruanlitao.shortlink.project.mq.redis.consumer;

import com.alibaba.fastjson2.JSON;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;
import top.ruanlitao.shortlink.project.common.convention.exception.ServiceException;
import top.ruanlitao.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import top.ruanlitao.shortlink.project.mq.idempotent.MessageQueueIdempotentHandler;
import top.ruanlitao.shortlink.project.service.ShortLinkStatsService;

import java.util.Map;
import java.util.Objects;

/**
 * 短链接监控状态保存消息队列消费者
 */
@ConditionalOnProperty(prefix = "short-link.message-queue", name = "type", havingValue = "redis")
@Slf4j
@Component
@RequiredArgsConstructor
public class RedisShortLinkStatsSaveConsumer implements StreamListener<String, MapRecord<String, String, String>> {

    private final StringRedisTemplate stringRedisTemplate;
    private final MessageQueueIdempotentHandler messageQueueIdempotentHandler;
    private final ShortLinkStatsService shortLinkStats;

    /**
     * 消费短链接状态消息
     */
    @Override
    public void onMessage(MapRecord<String, String, String> message) {
        // 读取消息获取统计数据
        String stream = message.getStream();
        RecordId id = message.getId();
        Map<String, String> producerMap = message.getValue();

        // 幂等性判断
        // 判断一下这个消息有没有被消费过
        if (messageQueueIdempotentHandler.isMessageBeingConsumed(id.toString())) {
            // 消费过了，再判断一下这个消息有没有消费完成
            if (messageQueueIdempotentHandler.isAccomplish(id.toString())) {
                // 消费完成，直接返回，不做任何处理了
                return;
            }
            // 消息没有消费完成，抛出异常，等待后续消息重发
            throw new ServiceException("消息未完成流程，需要消息队列重试");
        }
        // 消息没有被消费过，尝试消费这条消息
        try {
            ShortLinkStatsRecordDTO shortLinkStatsRecord = JSON.parseObject(producerMap.get("statsRecord"), ShortLinkStatsRecordDTO.class);
            log.info("短链接 {} 从消息队列中取出消费", shortLinkStatsRecord.getFullShortUrl());
            // 写入数据库
            shortLinkStats.actualSaveShortLinkStats(shortLinkStatsRecord);
            // 删除消息
            stringRedisTemplate.opsForStream().delete(Objects.requireNonNull(stream), id.getValue());
        } catch (Throwable ex) {
            // 消费消息的过程中发生异常，将幂等性判断记录删除掉，以便之后可以再次消费这条消息
            messageQueueIdempotentHandler.delMessageProcessed(id.toString());
            log.error("记录短链接监控消费异常", ex);
            throw ex;
        }
        // 消息被正常执行完成，设置这个消息为已完成状态
        messageQueueIdempotentHandler.setAccomplish(id.toString());
    }
}
